package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.CartAddUuBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class DwsTradeCartAddUuWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeCartAddUuWindow().start(Constant.TOPIC_DWD_TRADE_CART_ADD,Constant.DWS_TRADE_CART_ADD_UU_WINDOW,4,10026);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        // 1.读取数据源
        // 2.数据清洗
//        dataStreamSource.print();

        //etl
        SingleOutputStreamOperator<JSONObject> eltStream = eltStream(dataStreamSource);

//        eltStream.print();
        //3.添加水位线
        SingleOutputStreamOperator<JSONObject> waterStream = waterStream(eltStream);


        //4.keyby
        SingleOutputStreamOperator<CartAddUuBean> streamOperator = keyCartCV(waterStream);

//        streamOperator.print();

        // 5、开窗聚合
        SingleOutputStreamOperator<CartAddUuBean> windowStream = WindowStream(streamOperator);

//        reduce.print();

        // 6.写入到doris
        windowStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_TRADE_CART_ADD_UU_WINDOW));



    }

    private static SingleOutputStreamOperator<CartAddUuBean> WindowStream(SingleOutputStreamOperator<CartAddUuBean> streamOperator) {
        return streamOperator.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<CartAddUuBean>() {
                    @Override
                    public CartAddUuBean reduce(CartAddUuBean t1, CartAddUuBean t2) throws Exception {
                        t1.setCartAddUuCt(t1.getCartAddUuCt() + t2.getCartAddUuCt());

                        return t1;
                    }
                }, new AllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<CartAddUuBean> iterable, Collector<CartAddUuBean> collector) throws Exception {

                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<CartAddUuBean> iterator = iterable.iterator();

                        CartAddUuBean cartAddUuBean = iterator.next();

                        cartAddUuBean.setStt(DateFormatUtil.tsToDateTime(start));
                        cartAddUuBean.setEdt(DateFormatUtil.tsToDateTime(end));

                        cartAddUuBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));

                        collector.collect(cartAddUuBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<CartAddUuBean>  keyCartCV(SingleOutputStreamOperator<JSONObject> waterStream) {
        return waterStream.keyBy(x -> x.getString("user_id")).process(new ProcessFunction<JSONObject, CartAddUuBean>() {

            //计算 独立数
            ValueState<String> state;

            @Override
            public void open(Configuration parameters) {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.hours(24))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                state = getRuntimeContext().getState(stateDescriptor);

            }


            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, CartAddUuBean>.Context context, Collector<CartAddUuBean> collector) throws Exception {


                Long ts = jsonObject.getLong("ts");

                String tsToDate = DateFormatUtil.tsToDate(ts);

                String value = state.value();

                long cartcv = 0L;


                if (value == null || !value.equals(tsToDate)) {
                    cartcv = 1L;
                    state.update(tsToDate);
                }


                collector.collect(CartAddUuBean.builder()
                        .cartAddUuCt(cartcv)
                        .curDate(tsToDate)
                        .build());
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> waterStream(SingleOutputStreamOperator<JSONObject> eltStream) {
         return eltStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<JSONObject> eltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);
                    if (jsonObject != null) {

                        Long ts = jsonObject.getLong("ts");

                        String user_id = jsonObject.getString("user_id");

                        if (ts != null && user_id != null) {
                            jsonObject.put("ts", ts * 1000);

                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
